package com.zdb.demo.flink.test;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class ProcessFunctionDemo {

    public static void main(String[] args) {
        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
        streamEnv.setParallelism(4);

        streamEnv.socketTextStream("localhost", 9999).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value!=null && value.length() > 0;
            }
        }).map(new RichMapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                String[] strs = value.split(",");
                return new Tuple2<>(strs[0], strs[1]);
            }
        }).keyBy(0)
                .process(new CountWithTimeoutFunction())
                .print("process_fn").setParallelism(1);

        try {
            streamEnv.execute("stream process function");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class CountWithTimestamp {

    public String key;
    public int count;
    public long lastModified;

    public CountWithTimestamp() {
    }
}

class CountWithTimeoutFunction extends
        ProcessFunction<Tuple2<String, String>, Tuple2<String, Integer>> {

    /** process function维持的状态 */
    private ValueState<CountWithTimestamp> state;

    /** The state that is maintained by this process function */
    @Override
    public void open(Configuration parameters) throws Exception {
        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out)
            throws Exception {
        // get the state for the key that scheduled the timer
        //获取计划定时器的key的状态
        CountWithTimestamp result = state.value();

        // 检查是否是过时的定时器或最新的定时器
        if (timestamp == result.lastModified + 10000) {
            // emit the state on timeout
            out.collect(new Tuple2<String, Integer>(result.key, result.count));
        }
    }

    @Override
    public void processElement(Tuple2<String, String> value,
            Context ctx,
            Collector<Tuple2<String, Integer>> collector) throws Exception {
        // retrieve the current count
        // 获取当前的count
        CountWithTimestamp current = state.value();
        if (current == null) {
            current = new CountWithTimestamp();
            current.key = value.f0;
        }

        // update the state's count
        // 更新 state 的 count
        current.count++;

        // set the state's timestamp to the record's assigned event time timestamp
        // 将state的时间戳设置为记录的分配事件时间戳
        current.lastModified = ctx.timestamp();

        // write the state back
        // 将状态写回
        state.update(current);

        // schedule the next timer 60 seconds from the current event time
        // 从当前事件时间开始计划下一个60秒的定时器
        ctx.timerService().registerEventTimeTimer(current.lastModified + 10000);
    }
}
